package com.atguigu.hadoop.examples.mapreduce.terasort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.PureJavaCrc32;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.zip.Checksum;

public class TeraValidate extends Configured
        implements Tool
{
    private static final Text ERROR = new Text("error");
    private static final Text CHECKSUM = new Text("checksum");

    private static String textifyBytes(Text t) {
        BytesWritable b = new BytesWritable();
        b.set(t.getBytes(), 0, t.getLength());
        return b.toString();
    }

    private static void usage()
            throws IOException
    {
        System.err.println("teravalidate <out-dir> <report-dir>");
    }

    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf());
        if (args.length != 2) {
            usage();
            return 1;
        }
        TeraInputFormat.setInputPaths(job, new Path[] { new Path(args[0]) });
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.setJobName("TeraValidate");
        job.setJarByClass(TeraValidate.class);
        job.setMapperClass(ValidateMapper.class);
        job.setReducerClass(ValidateReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setNumReduceTasks(1);

        FileInputFormat.setMinInputSplitSize(job, 9223372036854775807L);
        job.setInputFormatClass(TeraInputFormat.class);
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args)
            throws Exception
    {
        int res = ToolRunner.run(new Configuration(), new TeraValidate(), args);
        System.exit(res);
    }

    static class ValidateReducer extends Reducer<Text, Text, Text, Text>
    {
        private boolean firstKey = true;
        private Text lastKey = new Text();
        private Text lastValue = new Text();

        public void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
            if (TeraValidate.ERROR.equals(key)) {
                for (Text val : values)
                    context.write(key, val);
            }
            else if (TeraValidate.CHECKSUM.equals(key)) {
                Unsigned16 tmp = new Unsigned16();
                Unsigned16 sum = new Unsigned16();
                for (Text val : values) {
                    tmp.set(val.toString());
                    sum.add(tmp);
                }
                context.write(TeraValidate.CHECKSUM, new Text(sum.toString()));
            } else {
                Text value = (Text)values.iterator().next();
                if (this.firstKey) {
                    this.firstKey = false;
                }
                else if (value.compareTo(this.lastValue) < 0) {
                    context.write(TeraValidate.ERROR, new Text("bad key partitioning:\n  file " + this.lastKey + " key " +
                            TeraValidate.textifyBytes(this.lastValue) +
                            "\n  file " + key + " key " +
                            TeraValidate.textifyBytes(value)));
                }

                this.lastKey.set(key);
                this.lastValue.set(value);
            }
        }
    }

    static class ValidateMapper extends Mapper<Text, Text, Text, Text>
    {
        private Text lastKey;
        private String filename;
        private Unsigned16 checksum = new Unsigned16();
        private Unsigned16 tmp = new Unsigned16();
        private Checksum crc32 = new PureJavaCrc32();

        private String getFilename(FileSplit split)
        {
            return split.getPath().getName();
        }

        public void map(Text key, Text value, Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException
        {
            if (this.lastKey == null) {
                FileSplit fs = (FileSplit)context.getInputSplit();
                this.filename = getFilename(fs);
                context.write(new Text(this.filename + ":begin"), key);
                this.lastKey = new Text();
            }
            else if (key.compareTo(this.lastKey) < 0) {
                context.write(TeraValidate.ERROR, new Text("misorder in " + this.filename + " between " +
                        TeraValidate.textifyBytes(this.lastKey) +
                        " and " +
                        TeraValidate.textifyBytes(key)));
            }

            this.crc32.reset();
            this.crc32.update(key.getBytes(), 0, key.getLength());
            this.crc32.update(value.getBytes(), 0, value.getLength());
            this.tmp.set(this.crc32.getValue());
            this.checksum.add(this.tmp);
            this.lastKey.set(key);
        }

        public void cleanup(Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException
        {
            if (this.lastKey != null) {
                context.write(new Text(this.filename + ":end"), this.lastKey);
                context.write(TeraValidate.CHECKSUM, new Text(this.checksum.toString()));
            }
        }
    }
}